package com.atguigu.cn.hotitems_analysis

import java.sql.Timestamp
import java.util.Properties

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * @author: yangShen
  * @Description: 获取实时热门商品，时间窗口->滑动窗口->( 固定窗口长度1个小时，滑动间隔5分钟 )
  * @Date: 2020/4/28 9:29 
  */
//定义输入数据的样例类
case class UserBehavior(userId: Long, itemId: Long, categoryId: Int, behavior: String, timestamp: Long)
//定义窗口输出的样例类
case class ItemWindowCount (itemId: Long, windowEnd: Long, count: Long)

object HotItems {
  def main(args: Array[String]): Unit = {

    //1.enviroment:创建执行环境
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度为1，乱序时方便查看数据
    environment.setParallelism(1)
    //设置时间语义
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //2.source:读取数据
    //2. 从kafka读取数据
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "10.16.26.16:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    val dataStream = environment.addSource(new FlinkKafkaConsumer[String]("hotitems", new SimpleStringSchema(), properties))

//    val dataStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")
      .map(data => {
        val dataArray = data.split(",")
        UserBehavior(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim.toInt, dataArray(3).trim, dataArray(4).trim.toLong)
      })
      //指定时间戳(参考WindowTest示例)，也可以指定waterMark(乱序时)：引用的UserBehavior.csv中时间戳为升序，比较方便 可以直接指定时间戳即可
      .assignAscendingTimestamps(_.timestamp * 1000L)

    //3.transform 处理数据
    val processStream = dataStream
        .filter(_.behavior == "pv")
        .keyBy(_.itemId)
        .timeWindow( Time.hours(1), Time.minutes(5) )
        //参数(1.preAggregator：预聚合，来一个处理一个，把状态聚合出来；2.windowFunction：用windowFunction输出包装好的结果),第一个函数的输出是第二个函数的输入
        .aggregate(new CountAgg(), new WindowResult())  //窗口聚合
        .keyBy(_.windowEnd)   //按照窗口进行分组
        .process(new TopNumberHotItems(3))

    //4.sink:控制台输出
    processStream.print()

    environment.execute("hot items job")
  }

}

//自定义预聚和函数    ---参数(输入参数UserBehavior, 计数器 初始为0, 输出计数的统计)
class CountAgg() extends AggregateFunction[UserBehavior, Long, Long] {
  override def createAccumulator(): Long = 0

  override def add(value: UserBehavior, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

//自定义预聚和函数 计算平均数: 计算时间戳timestamp平均值      ---参数(输入参数UserBehavior, 计数器(时间戳, 次数), 输出时间戳平均值)
class AverageAgg() extends AggregateFunction[UserBehavior, (Long, Int), Double] {
  //(时间戳，时间戳出现的次数)
  override def createAccumulator(): (Long, Int) = (0L, 0)
  //(每一的时间戳相加(在上次的基础之上)，次数+1)
  override def add(value: UserBehavior, accumulator: (Long, Int)): (Long, Int) = (accumulator._1 + value.timestamp, accumulator._2 + 1)

  override def getResult(accumulator: (Long, Int)): Double = accumulator._1 / accumulator._2

  override def merge(a: (Long, Int), b: (Long, Int)): (Long, Int) = (a._1 + b._1, a._2 + b._2)
}

//自定义窗口函数，输出ItemViewCount    ---参数(输入的是CountAgg()的输出，输出ItemWindowCount，key是itemId(来源于前面的key: .keyBy(_.itemId)), 窗口类型)
class WindowResult() extends WindowFunction[Long, ItemWindowCount, Long, TimeWindow] {
  override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemWindowCount]): Unit = {
    out.collect(ItemWindowCount(key, window.getEnd, input.iterator.next()))
  }
}

//自定义的处理函数    ---参数(key是itemId, 输入是ItemWindowCount, 输出)
class TopNumberHotItems(topSize: Int) extends KeyedProcessFunction[Long, ItemWindowCount, String]{

  private var itemStates: ListState[ItemWindowCount] = _

  override def open(parameters: Configuration): Unit = {
    itemStates = getRuntimeContext.getListState(new ListStateDescriptor[ItemWindowCount]("item-state", classOf[ItemWindowCount]))
  }

  override def processElement(value: ItemWindowCount, ctx: KeyedProcessFunction[Long, ItemWindowCount, String]#Context, out: Collector[String]): Unit = {
    //把每条数据存入状态列表
    itemStates.add(value)

    //注册一个定时器，延迟1毫秒触发,当触发时说明收齐了属于windowEnd窗口的所有数据
    //也就是说，当程序看到windowEnd + 1 的水位线waterMark时，触发onTimer回调函数
    ctx.timerService().registerEventTimeTimer( value.windowEnd + 1 )
  }

  //定时器触发时，对所有数据排序，并输出结果
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemWindowCount, String]#OnTimerContext, out: Collector[String]): Unit = {
    //将所有state中的数据取出，放到一个ListBuffer中
    var allItems: ListBuffer[ItemWindowCount] = new ListBuffer()
    //将itemStates遍历循环放到listBuffer中
    import scala.collection.JavaConversions._
    for(item <- itemStates.get()){
      allItems += item
    }

    //按照count大小排序，降序排列，并取前n个 = topSize
    val sortItems = allItems.sortBy(_.count)(Ordering.Long.reverse).take(topSize)

    //清空状态
    itemStates.clear()

    //将排名结果格式化输出
    val result: StringBuilder = new StringBuilder()
    //定时器+1，这里要-1取原值，用窗口关闭的事件
    result.append("时间：").append(new Timestamp(timestamp - 1)).append("\n")

    //输出每一个商品信息
    for(i <- sortItems.indices){
      val currentItem = sortItems(i)
      result.append("No").append(i + 1).append(":").append("商品ID=").append(currentItem.itemId).append("，浏览量=").append(currentItem.count).append("\n")

    }

    result.append("=======================================")

    //控制输出频率
    Thread.sleep(1000)
    out.collect(result.toString())
  }
}
